package com.yjd.bi.canal.service.impl;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.yjd.bi.canal.dao.ICanalMessageDao;
import com.yjd.bi.canal.dao.impl.CanalMessageDaoImpl;
import com.yjd.bi.canal.model.CanalMessageModel;
import com.yjd.bi.canal.model.Entrys;
import com.yjd.bi.canal.model.Records;
import com.yjd.bi.canal.service.ICanalMessageService;
import com.yjd.bi.report.model.DataChangeRecordModel;
import com.yjd.comm.base.dao.IBaseDao;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.comm.base.service.impl.BaseServiceImpl;
import com.yjd.comm.util.FrameUtil;
import com.yjd.comm.util.StringUtil;

/**
 * <b>description</b>：客户消息订单业务实现<br>
 * <b>time</b>：2016-07-13 15:14:27 <br>
 * <b>author</b>：  ready percy-chen@hotmail.com
 */
@Service("canalMessageService")
public class CanalMessageServiceImpl extends BaseServiceImpl implements ICanalMessageService {

	/**
	 * 插入
	 * 
	 * @param model
	 * @return
	 * @throws Exception
	 */
	public CanalMessageModel insert(CanalMessageModel model) throws Exception {
		return this.canalMessageDao.insertModel(model, true);
	}
	
	public void inserBatch(List<CanalMessageModel> list) throws Exception{
		this.canalMessageDao.insertBatch(list, true);
	}

	/**
	 * 更新
	 * 
	 * @param model
	 * @return
	 * @throws Exception
	 */
	public int update(CanalMessageModel model) throws Exception {
		return this.canalMessageDao.updateModel(model, true);
	}
	
	

	/**
	 * 根据id删除
	 * 
	 * @param id
	 * @return
	 * @throws Exception
	 */
	public int delete(long id) throws Exception {
		Map<Object, Object> paramMap = new HashMap<Object, Object>();
		paramMap.put("id", id);
		return this.canalMessageDao.deleteModel(paramMap, true);
	}
	
	public int updateByCanalMessage(String table,Records keymap,List<Records> changeList) throws Exception{
		return ((CanalMessageDaoImpl)this.canalMessageDao).updateByMessage(FrameUtil.newHashMap("table_name",table,"keymap",keymap,"changeList",changeList), true);
	}
	
	public Map insertByCanalMessage(String table,List<String> fieldList,List<String> valueList) throws Exception{
		return ((CanalMessageDaoImpl)this.canalMessageDao).insertByMessage(FrameUtil.newHashMap("table_name",table,"fieldList",fieldList,"valueList",valueList), true);
	}
	
	public CanalMessageModel insertLog(CanalMessageModel model) throws Exception{
		return ((CanalMessageDaoImpl)this.canalMessageDao).insertLog(model, true);
	}
	
	public int deleteByCanalMessage(String table,Records keymap) throws Exception{
		return ((CanalMessageDaoImpl)this.canalMessageDao).deleteByMessage(FrameUtil.newHashMap("table_name",table,"keymap",keymap), true);
	}
	
	public void moveToLog(List<CanalMessageModel> ms) throws Exception{
		List<Long> idList=FrameUtil.newArrayList();
		for(CanalMessageModel m:ms){
		if(m.getId()!=null&&m.getStatus().intValue()==1){
			this.insertLog(m);
			idList.add(m.getId());
		}}
		this.canalMessageDao.deleteModel(FrameUtil.newHashMap("idList",idList), true);
	}
	
	public boolean transfer(List<CanalMessageModel> list) throws Exception{
		List<DataChangeRecordModel> changelogs=FrameUtil.newArrayList();
		for(CanalMessageModel cm:list){
			switch (cm.getEvent_type()) {
				case EventType.ALTER_VALUE :{
					String sql=cm.getEntrydata().replaceAll(cm.getSchema_name(),"bi_yijiedai_com");
					 ((CanalMessageDaoImpl)this.canalMessageDao).excuteSql(sql, true);
					break;
				}
				case EventType.CINDEX_VALUE :{
					break;
				}
				case EventType.DINDEX_VALUE:{
					break;
				}
				case EventType.ERASE_VALUE:{
					break;
				}
				case EventType.INSERT_VALUE :
					{
						List<Entrys> elist=JSON.parseArray(cm.getEntrydata(), Entrys.class);
						Records keymap=new Records();
						List<String> fieldList=new ArrayList<>();
						List<String> valueList=new ArrayList<>();
						for(Entrys e:elist){
							if(StringUtil.isNotEmpty(e.getValue_new())){
								fieldList.add(e.getField());
								valueList.add(e.getValue_new());
							}
							if(e.getIskey().intValue()==1){
								keymap.setField(e.getField());
								keymap.setValue(e.getValue_new());
							}
						}
						if(!keymap.isEmpty()){
							deleteByCanalMessage(cm.getTable_name(), keymap);
						}
						if(fieldList.size()>0&&valueList.size()>0){
							insertByCanalMessage(cm.getTable_name(), fieldList, valueList);
						}
						break;
					}
				case EventType.UPDATE_VALUE :{
					List<Entrys> elist=JSON.parseArray(cm.getEntrydata(), Entrys.class);
					Records keymap=new Records();
					List<Records> changeList=FrameUtil.newArrayList();
					for(Entrys e:elist){
						if(e.getIskey()!=null&&e.getIskey().intValue()==1){
							keymap.setField(e.getField());
							keymap.setValue(e.getValue_new());
						}
						if(e.getUpdate().intValue()==1){
							changeList.add(new Records(e.getField(),e.getValue_new()));
						}
					}
					if(!keymap.isEmpty()&&changeList.size()>0){
						updateByCanalMessage(cm.getTable_name(), keymap, changeList);
					}
					break;
				}
				case EventType.DELETE_VALUE:{
					List<Entrys> elist=JSON.parseArray(cm.getEntrydata(), Entrys.class);
					Records keymap=new Records();
					for(Entrys e:elist){
						if(e.getIskey().intValue()==1){
							keymap.setField(e.getField());
							keymap.setValue(e.getValue_old());
						}
					}
					if(!keymap.isEmpty()){
						deleteByCanalMessage(cm.getTable_name(), keymap);
					}
					break;
				}
				default :
					break;
			}
			cm.setStatus(1);;
			this.canalMessageDao.updateByMap(FrameUtil.newHashMap("id",cm.getId(),"status",1), true) ;
		}
		return true;
	}

	public List getTableList(DbWREnums dbwr)
			throws Exception {
		return ((CanalMessageDaoImpl)this.canalMessageDao).getTableList(dbwr.isWrite());
	}
	
	@Resource
	private ICanalMessageDao canalMessageDao;

	@Override
	public IBaseDao getBaseDao() {
		return this.canalMessageDao;
	}

}
